package com.example.spring.pulsar.demo;

import com.example.spring.pulsar.demo.domain.so.UserSo;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.schema.JSONSchema;

import java.nio.charset.StandardCharsets;
import java.time.LocalDate;


/**
 * @description: pulsar 数据的产生和消费
 * @author: 吴佳伟
 * @date: 2021/5/20
 */

/**
 * 消费者
 */
class PulsarConsumer {

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.17.221:30718")
                .build();

        Consumer consumer = client.newConsumer(JSONSchema.of(UserSo.class))
                .topic("UserSo-topic")
                .subscriptionName("UserSo-subscription")
                .subscribe();

        while (true) {
            // Wait for a message
            Message msg = consumer.receive();

            try {
                MessageImpl message= (MessageImpl) msg;

                // Do something with the message
                System.out.printf("schema : %s", message.getSchema().getSchemaInfo());
                System.out.printf("Message received: %s", new String(message.getData()));

                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
            } catch (Exception e) {
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }
        }


    }
}


/**
 * 消息提供者
 */
class PulsarProvider {
    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.17.221:30718")
                .build();
        Producer<UserSo> producer = client.newProducer(JSONSchema.of(UserSo.class))
                .topic("UserSo-topic").create();

        while (true) {
            UserSo userSo = new UserSo();
            userSo.setAddress("测试地址");
            userSo.setName("测试人员");
            userSo.setAge(1000L);
            producer.send(userSo);
        }
    }
}

